﻿using Basic.EventBus.Attributes;
using Basic.EventBus.Eventbus;
using Confluent.Kafka;
using System.Reflection;

namespace Basic.EventBus.EventBusKafka;

public class KafkaSubscriber : IEventSubscriber
{
    private KafkaBusBrokerConfig Config;
    private List<KafkaConsumerRecord> RecordList;

    public KafkaSubscriber(KafkaBusBrokerConfig options)
    {
        Config = options;
        RecordList = new List<KafkaConsumerRecord>(8);
    }

    public void AddSubscriber(ISubscriberHandler subscriberInfo)
    {
        var attrObj = subscriberInfo.GetType().GetCustomAttribute<EventSubscriberAttribute>();
        if (attrObj == null) return;

        var record = new KafkaConsumerRecord(subscriberInfo, Config);
        record.Start(attrObj);
        RecordList.Add(record);
    }

    public void AddSubscriber(ISubscriberHandler[] subscriberInfos)
    {
        if (subscriberInfos?.Length > 0)
            foreach (var item in subscriberInfos)
                AddSubscriber(item);
    }

    #region 消费逻辑

    internal class KafkaConsumerRecord
    {
        private SemaphoreSlim ConcurrentLock;
        private KafkaBusBrokerConfig Config;
        private EventSubscriberAttribute SubscriberAttributeObj;

        public IConsumer<string, byte[]> Consumer { get; private set; }
        public ISubscriberHandler Subscriber { get; private set; }

        public KafkaConsumerRecord(ISubscriberHandler handler, KafkaBusBrokerConfig config)
        {
            Subscriber = handler;
            Config = config;
        }

        private IConsumer<string, byte[]> CreateConsumerHanlder()
        {
            ConsumerConfig conf = new ConsumerConfig()
            {
                BootstrapServers = Config.Servers,
                GroupId = SubscriberAttributeObj.TopicObj.Group,
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnableAutoCommit = SubscriberAttributeObj.AutoAck,
            };
            var builder = new ConsumerBuilder<string, byte[]>(conf);
            return builder.Build();
        }

        public void Start(EventSubscriberAttribute subscriberAttribute)
        {
            SubscriberAttributeObj = subscriberAttribute;
            ConcurrentLock = new SemaphoreSlim(1, SubscriberAttributeObj.ConcurrencyNumber);

            Consumer = CreateConsumerHanlder();
            Consumer.Subscribe(SubscriberAttributeObj.TopicObj.Topic);

            _ = Task.Factory.StartNew(async () =>
            {
                while (Subscriber.TokenSource.Token.CanBeCanceled)
                {
                    await ReceiveMessage();
                }
                Consumer.Dispose();
                Consumer = null;
            }, Subscriber.TokenSource.Token);
        }

        private async Task ReceiveMessage()
        {
            try
            {
                await ConcurrentLock.WaitAsync();

                var result = Consumer.Consume();
                _ = Task.Factory.StartNew(async obj =>
                {
                    await Subscriber.HandleMessage(obj as byte[]);
                    ConcurrentLock.Release();
                }, result.Message.Value);
                if (!SubscriberAttributeObj.AutoAck)
                    Consumer.Commit(result);
            }
            catch (Exception ex)
            {

            }
        }

    }

    #endregion


    #region Dispose
    private bool disposedValue;

    protected virtual void Dispose(bool disposing)
    {
        if (!disposedValue)
        {
            if (disposing)
            {
                // TODO: 释放托管状态(托管对象)
            }
            // TODO: 将大型字段设置为 null
            disposedValue = true;
        }
    }

    public void Dispose()
    {
        // 不要更改此代码。请将清理代码放入“Dispose(bool disposing)”方法中
        Dispose(disposing: true);
        GC.SuppressFinalize(this);
    }
    #endregion

}